#include "app_pool.h"
#include <mqueue.h>
#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include "log/log.h"

#define MAX_MQ_NUM 10   // 消息队列最大数量

static char *mq_name = "/mq_pool";
static mqd_t mqd;
static int pool_count = 0;
static pthread_t *pool_thread;

void *App_Pool_Thread(void *arg)
{
    Task task;
    while(1)
    {
        // 从消息队列中取出任务
        ssize_t len = mq_receive(mqd, (char *)&task, sizeof(Task), NULL);
        if (len == sizeof(Task))
        {
            // 执行任务
            task.Task_func(task.arg);
        }
    }
}

int App_Pool_Init(int thread_num)
{
    struct mq_attr attr;
    attr.mq_maxmsg = MAX_MQ_NUM;
    attr.mq_msgsize = sizeof(Task);
    // 创建消息队列
    mqd = mq_open(mq_name, O_CREAT | O_RDWR, 0644, &attr);
    if (mqd == -1)
    {
        log_error("mq_open error");
        return -1;
    }
    // 创建线程池
    pool_count = thread_num;
    pool_thread = (pthread_t*)malloc(sizeof(pthread_t) * thread_num);
    for (int i = 0; i < thread_num; i++)
    {
        pthread_create(&pool_thread[i], NULL, App_Pool_Thread, NULL);
    }
    return 0;
}

void App_Pool_close(void)
{
    mq_close(mqd);
    mq_unlink(mq_name);
    for (int i = 0; i < pool_count; i++)
    {
        pthread_cancel(pool_thread[i]);
        pthread_join(pool_thread[i], NULL);
    }
    free(pool_thread);
}

void App_Pool_RegisterTask(int (*Task_func)(void *), void *arg)
{
    Task task;
    task.Task_func = Task_func;
    task.arg = arg;
    mq_send(mqd, (char *)&task, sizeof(Task), 1);
}
